package tt.dz.tcp.socket;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import javax.annotation.PostConstruct;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.replay.ReplayingDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import tt.dz.dao.ext.charger.OmsChargerDao;
import tt.dz.dao.ext.dic.OmsDicForwardSrvInfoDao;
import tt.dz.entity.system.InstructionLog;
import tt.dz.service.imp.middleware.ev3c2MessageServiceImpl;
import tt.dz.service.imp.middleware.instruction.Ev3c2Instructions;
import tt.dz.service.imp.middleware.vo.ProtocolVersion;
import tt.dz.service.sys.InstructionLogService;
import tt.dz.tcp.socket.exception.CreateConnectionFailException;
import tt.dz.tcp.socket.exception.DuplicateInstructionException;
import tt.dz.tcp.socket.exception.RetryFailException;
import tt.dz.util.StringUtil;

/**
 * 中间件通讯长连接类
 */
@Component
public class LongConnectionSendMsgImpl implements SendMsg
{
	private static HashMap<String, MessageFuture> messageFutureMap = new HashMap<>(); // MessageFuture对象map,每个message一个Future,收到回复或者超时将删除该Future

	@Autowired
	OmsChargerDao omsChargerDao;

	@Autowired
	OmsDicForwardSrvInfoDao omsDicForwardSrvInfoDao;

	@Autowired
	private ReceiveEventListener receiveListener;

	@Autowired
	private MessageStorage messageStorage;

	@Autowired
	InstructionLogService insLogDaoService;
	
	private ReadWriteLock lock = new ReentrantReadWriteLock();

	private ConcurrentHashMap<Integer, Channel> closedChannels = new ConcurrentHashMap<>();

	private Set<Integer> activeChannels = new HashSet<>();//目前存活的channel集合

	private Log log = LogFactory.getLog(LongConnectionSendMsgImpl.class);

	//	@Value(value = "${dz.tcp.client.connectionPoolSize}")
	private static final int connectionPoolSize = 10;

	//For processing received messages
	private final ExecutorService receiveExecutorService = Executors.newCachedThreadPool();//消息接收处理线程组

	private Map<Integer, DzConnectionPool> group_connectionpool = new HashMap<>();

	private ClientBootstrap bootstrap = null;

	//	private Jedis jedis;

	@PostConstruct
	public void init()
	{
		log.debug("Init");
		ChannelFactory factory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());

		bootstrap = new ClientBootstrap(factory);

		bootstrap.setPipelineFactory(new ChannelPipelineFactory()
		{
			public ChannelPipeline getPipeline()
			{
				ChannelPipeline pipeline = Channels.pipeline();
				pipeline.addLast("decoder", new DzLongConnectionFrameDecoder());
				//pipeline.addLast("decoder2", new StringDecoder(Charset.forName("utf-8")));
				pipeline.addLast("encoder", new StringEncoder(Charset.forName("utf-8")));
				pipeline.addLast("handler", new DzLongConnectionClientHandler());
				return pipeline;
			}
		});

		bootstrap.setOption("tcpNoDelay", true);
		bootstrap.setOption("keepAlive", true);
		bootstrap.setOption("connectTimeoutMillis", 10000l);

		createConnectionPool();
	}

	/**
	 * 创建连接池
	 */
	private void createConnectionPool()
	{
		List<MiddleWareServer> list_middleware = getRemoteServer();
		log.info(connectionPoolSize);
		for (MiddleWareServer server : list_middleware)
		{
			DzConnectionPool pool = new DzConnectionPool();

			pool.ip = server.serverIp;
			pool.port = server.serverPort;
			pool.protocolVersion = server.protocolVersion;
			log.debug(pool.ip);
			log.debug(pool.port);
			group_connectionpool.put(server.serverId, pool);
			for (int i = 0; i < connectionPoolSize; i++)
			{
				Channel channel = createConnection(pool.ip, pool.port);
				if (channel != null)
				{
					Lock writelock = lock.writeLock();
					try
					{
						writelock.lock();
						DzConnection dzConnection = new DzConnection(channel);
						pool.connectionPool[i] = dzConnection;
						activeChannels.add(channel.getId());
					}
					catch (Exception e)
					{
						log.error(e.getMessage(), e);
						throw e;
					}
					finally
					{
						writelock.unlock();
					}
				}
			}
		}
	}

	/**
	 * 查询所有中间件服务器
	 */
	private List<MiddleWareServer> getRemoteServer()
	{
		List<Map<String, Object>> list_server = omsDicForwardSrvInfoDao
				.findDataBySql("SELECT s.forward_srv_id, s.main_hardware_ip, s.main_port_up, s.protocol_version FROM oms_dic_forward_srv_info s WHERE s.valid_flag = 1");
		List<MiddleWareServer> list = new ArrayList<>();
		for (Map<String, Object> server : list_server)
		{
			MiddleWareServer s = new MiddleWareServer();
			s.serverId = (Integer) server.get("forward_srv_id");
			s.serverIp = (String) server.get("main_hardware_ip");
			s.serverPort = (Integer) server.get("main_port_up");
			String protocol_version = (String) server.get("protocol_version");
			if ("0101".equals(protocol_version) || "0102".equals(protocol_version))
				s.protocolVersion = ProtocolVersion.Version_1;
			else if ("0201".equals(protocol_version))
				s.protocolVersion = ProtocolVersion.Version_2;
			list.add(s);
		}
		return list;
	}

	/**
	 * 根据桩号查询其对应连接上的中间件服务器
	 * @param chargerid 桩号
	 * @return 中间件服务器编号
	 */
	private Integer getServerId(String chargerid)
	{
		Integer serverId = null;
		List<Map<String, Object>> chargers = omsChargerDao.findDataBySql("select c.forward_srv_id from oms_charger c where c.charger_id = ?", chargerid);
		if (!chargers.isEmpty())
		{
			serverId = (Integer) chargers.get(0).get("forward_srv_id");
		}
		return serverId;
	}

	/**
	 * 创建连接
	 */
	private Channel createConnection(String ip, int port)
	{
		log.info(String.format("try to create connection with ip: %1$s, port: %2$d", ip, port));
		ChannelFuture cf = bootstrap.connect(new InetSocketAddress(ip, port));
		cf.awaitUninterruptibly();
		Channel channel = null;
		if (cf.isSuccess())
		{
			channel = cf.getChannel();
			log.debug(String.format("Connection create successful, Channel Id: %1$d, ip: %2$s, port: %3$d", channel.getId(), ip, port));
			log.debug(channel);
		}
		else
		{
			log.error("Connection create failed");
		}
		return channel;
	}

	/**
	 * 判断是否需要启动断线重连
	 */
	public boolean needReconnect()
	{
		Lock readlock = lock.readLock();
		try
		{
			readlock.lock();
			return !closedChannels.isEmpty();
		}
		catch (Exception e)
		{
			log.error(e.getMessage(), e);
			throw e;
		}
		finally
		{
			readlock.unlock();
		}
	}

	/**
	 * 重新创建连接,
	 */
	public void reConnect()
	{
		log.info("Re-connecting");
		Lock writelock = lock.writeLock();

		//当有必要重新连接时, closedChannels集合里会有之前被中断的channel对象
		//将该channel对象从connectionPool数组中移除, 然后创建新的channel, 添加进connectionPool数组中, 并同时更新存活channel集合activeChannels
		try
		{
			writelock.lock();
			log.debug(closedChannels);
			for (Integer closed_channel_id : closedChannels.keySet())
			{
				log.info(closed_channel_id);
				log.debug(closedChannels.keySet().size());

				Channel closed_channel = closedChannels.get(closed_channel_id);
				log.info(closed_channel);

				int index_inpool = -1;

				DzConnectionPool pool = null;

				boolean found = false;
				for (DzConnectionPool p : group_connectionpool.values())
				{
					for (int i = 0; i < p.connectionPool.length; i++)
					{
						if (p.connectionPool[i].getChannel() == closed_channel)
						{
							log.info("found channel in pool");
							index_inpool = i;
							pool = p;
							found = true;
							break;
						}
					}
				}

				log.info(index_inpool);
				log.info(pool);

				if (found)
				{
					Channel new_channel = createConnection(pool.ip, pool.port);
					if (new_channel != null)
					{
						log.info(new_channel);
						DzConnection dz = new DzConnection(new_channel);
						pool.connectionPool[index_inpool] = dz;
						closedChannels.remove(closed_channel_id);
						activeChannels.remove(closed_channel_id);
						activeChannels.add(new_channel.getId());
					}
				}
				else
				{
					closedChannels.remove(closed_channel_id);
				}
				log.info("loop end");
			}
		}
		catch (Exception e)
		{
			log.error(e.getMessage(), e);
			throw e;
		}
		finally
		{
			writelock.unlock();
		}

	}

	private DzConnection nextConnection(String chargerid) throws CreateConnectionFailException
	{
		Integer serverId = getServerId(chargerid);
		DzConnectionPool pool = group_connectionpool.get(serverId);
		if (pool == null)
			throw new CreateConnectionFailException("Cannot map to charger id to a existing middlerware ");

		int lastIndex = connectionPoolSize - 1;
		if (lastIndex == 0)
			return pool.connectionPool[0];

		int currentValue = pool.currentIndex.get();
		if (currentValue == lastIndex)
		{
			pool.currentIndex.compareAndSet(currentValue, 0);
			return pool.connectionPool[lastIndex];
		}
		if (currentValue > lastIndex)
		{
			pool.currentIndex.compareAndSet(currentValue, 0);
			return nextConnection(chargerid);
		}
		else
		{
			pool.currentIndex.getAndIncrement();
			return pool.connectionPool[currentValue];
		}
	}

	/**
	 * 消息发送器, 找到下一个可用的连接并发送消息
	 */
	private void sendByCount(final String msg, final String chargerid, final int count) throws CreateConnectionFailException, RetryFailException
	{
		if (count == 0)
		{
			log.error("failed after trying " + connectionPoolSize + "times");
			throw new RetryFailException();
		}

		final DzConnection dzConnection = nextConnection(chargerid);
		final Channel channel = dzConnection.getChannel();

		if (channel == null)
			throw new CreateConnectionFailException("cannot get a sending channel");

		final AtomicInteger usingCount = dzConnection.getUsingCount();
		int currentUsingCount = usingCount.get();
		if (currentUsingCount == -1)
		{
			sendByCount(msg, chargerid, count - 1);
			return;
		}
		if (usingCount.compareAndSet(currentUsingCount, currentUsingCount + 1))
		{
			if (usingCount.get() == -1)
			{
				sendByCount(msg, chargerid, count - 1);
				return;
			}

			if (!channel.isWritable())
			{
				//表示该连接当前无法写入数据, 需要使用下一个连接
				usingCount.getAndDecrement();
				sendByCount(msg, chargerid, count - 1);
				return;
			}
			else
			{
				ChannelFuture cf = channel.write(msg);
				cf.addListener(new ChannelFutureListener()
				{
					@Override
					public void operationComplete(ChannelFuture future) throws Exception
					{
						if (!future.isSuccess())
						{
							//							reConnect(channel);
							sendByCount(msg, chargerid, count - 1);
							return;
						}
						else
						{
							usingCount.getAndDecrement();
							log.debug("Send message successful, message is: " + msg);
							return;
						}
					}
				});
			}
		}
		else
		{
			sendByCount(msg, chargerid, count - 1);
			return;
		}
	}

	@Override
	public MessageFuture sendAsync(String msg, String chargerId) throws CreateConnectionFailException, RetryFailException, DuplicateInstructionException
	{
		if (msg == null)
			throw new NullPointerException("Message is empty");

		if (!msg.startsWith("data="))
			msg = "data=".concat(StringUtil.complement(String.valueOf(msg.length() + 9), 4)).concat(msg);

		//每次发送消息, 会把该条消息保存在redis中, 作为之后的重发和消息接收匹配用
		if (messageStorage != null)
			messageStorage.store(msg, chargerId);
		try {
//			writeLog(msg,chargerId);//pengyu add
			insLogDaoService.sendMsg(msg, 1, chargerId);
		} catch (Exception e) {
			log.error(e.getMessage());
			e.printStackTrace();
		}
		log.info("after writeLog ZhuDongSendToCharger");
		sendByCount(msg, chargerId, connectionPoolSize);
		
		String originMsg = msg.substring(9);
		MessageFuture future = new DefaultMessageFuture(originMsg);
		messageFutureMap.put(originMsg, future); //MessageFuture将会返回给调用方, 调用方使用 MessageFuture可以追踪消息是否成功返回, 错误响应码等等

		return future;
	}


	
	/**
	 * 定时任务将使用该方法重发报文
	 */
	public void resend(String msg, String chargerId) throws RetryFailException, CreateConnectionFailException
	{
		if (msg == null)
			throw new NullPointerException("Message is empty");

		if (!msg.startsWith("data="))
			msg = "data=".concat(StringUtil.complement(String.valueOf(msg.length() + 9), 4)).concat(msg);

		sendByCount(msg, chargerId, connectionPoolSize);
	}

	public static HashMap<String, MessageFuture> getMessageFutureMap()
	{
		return messageFutureMap;
	}

	private static class MiddleWareServer
	{
		private int serverId;
		private String serverIp;
		private int serverPort;
		private ProtocolVersion protocolVersion;
	}

	/**
	 * wrapped channel object by introducing a state flag
	 */
	private static class DzConnection
	{
		private Channel channel = null;

		/**
		 * current count of using this channel, initialize as 0, -1 stands for needing re-connect
		 */
		private AtomicInteger usingCount = new AtomicInteger(0);

		public DzConnection(Channel channel)
		{
			this.channel = channel;
		}

		public Channel getChannel()
		{
			return channel;
		}

		public AtomicInteger getUsingCount()
		{
			return usingCount;
		}
	}

	private static class DzConnectionPool
	{
		private ProtocolVersion protocolVersion;

		private DzConnection[] connectionPool = new DzConnection[connectionPoolSize];

		//		private final Map<Integer, DzConnection> channelMap = new ConcurrentHashMap<Integer, DzConnection>();

		private final AtomicInteger currentIndex = new AtomicInteger(0);

		private String ip;

		private int port;

		public ProtocolVersion getProtocolVersion()
		{
			return protocolVersion;
		}

		public void setProtocolVersion(ProtocolVersion protocolVersion)
		{
			this.protocolVersion = protocolVersion;
		}

		public DzConnection[] getConnectionPool()
		{
			return connectionPool;
		}

		public void setConnectionPool(DzConnection[] connectionPool)
		{
			this.connectionPool = connectionPool;
		}

		public String getIp()
		{
			return ip;
		}

		public void setIp(String ip)
		{
			this.ip = ip;
		}

		public int getPort()
		{
			return port;
		}

		public void setPort(int port)
		{
			this.port = port;
		}

		public AtomicInteger getCurrentIndex()
		{
			return currentIndex;
		}

	}

	public enum DzLongConnectionDecoderState
	{
		READ_HEADER,
		READ_CONTENT;
	}

	public static class DzLongConnectionFrameDecoder extends ReplayingDecoder<DzLongConnectionDecoderState>
	{
		private int length;

		public DzLongConnectionFrameDecoder()
		{
			super(DzLongConnectionDecoderState.READ_HEADER);
		}

		@Override
		protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buf, DzLongConnectionDecoderState state) throws Exception
		{
			switch (state)
			{
				case READ_HEADER:
					String header1Str = new String(buf.readBytes(4).array(), "UTF-8");
					length = Integer.valueOf(header1Str).intValue();
					checkpoint(DzLongConnectionDecoderState.READ_CONTENT);
				case READ_CONTENT:
					String content = new String(buf.readBytes(length).array(), "UTF-8");
					checkpoint(DzLongConnectionDecoderState.READ_HEADER);
					return content;
				default:
					throw new Error("Shouldn't reach here.");
			}
		}
	}

	/**
	 * 消息接收handler
	 */
	private class DzLongConnectionClientHandler extends SimpleChannelUpstreamHandler
	{
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception
		{
			final String msg = (String) e.getMessage();
			log.debug("message received " + msg);
			if (receiveListener != null)
			{
				final Channel channel = ctx.getChannel();
				receiveExecutorService.submit(new Runnable()
				{
					@Override
					public void run()
					{
						receiveListener.onReceiveMessage(msg, channel);
					}
				});
			}
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception
		{
			super.exceptionCaught(ctx, e);
			log.error(e.getChannel());

			log.error("Exception caught", e.getCause());
		}

		/**
		 * 重写该方法可以监听channel的close事件, 当连接中断时, 将其加入closedChannels集合
		 */
		@Override
		public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception
		{
			super.channelClosed(ctx, e);
			log.debug("Closed");
			Channel channel = e.getChannel();

			Lock readlock = lock.readLock();
			try
			{
				readlock.lock();
				if (!activeChannels.contains(channel.getId()))
					return;
			}
			catch (Exception ex)
			{
				log.error(ex.getMessage(), ex);
				throw ex;
			}
			finally
			{
				readlock.unlock();
			}

			Lock writelock = lock.writeLock();
			try
			{
				writelock.lock();
				if (closedChannels.containsKey(channel.getId()))
				{
					log.debug(String.format("channel with id: (%1$s) already exists in closed map", channel.getId()));
					return;
				}
				closedChannels.put(channel.getId(), channel);
			}
			catch (Exception ex)
			{
				log.error(ex.getMessage(), ex);
				throw ex;
			}
			finally
			{
				writelock.unlock();
			}
		}
	}
}
